package top.xphu.config;

import lombok.Data;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import top.xphu.commom.ResourceUtils;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author xphu
 * @version 1.0
 * @date 2021/9/3 12:18
 */
@Data
@Component
public class KafkaConfig implements BeanPostProcessor {

    private final KafkaProducer<String, String> producer;
    private final KafkaConfigProperties configProperties;

    private KafkaConfig(KafkaConfigProperties configProperties) {
        this.configProperties = configProperties;
        Properties props = new Properties();
//        1、安全认证相关配置
        System.setProperty("java.security.auth.login.config", ResourceUtils.build().getResourceFilePath("kafka/jaas.conf"));
        System.setProperty("java.security.krb5.conf", ResourceUtils.build().getResourceFilePath("kafka/krb5.conf"));
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put("sasl.kerberos.service.name", "kafka");
        props.put("sasl.mechanism", "GSSAPI");
//      2、producer基本配置
        props.put("bootstrap.servers", "tdh02:9092,tdh03:9092,tdh04:9092");
        props.put("acks", "1");//leader收到即可
        props.put("retries", 5);//重试次数，默认0
        props.put("retry.backoff.ms", 500);//重试时间间隔,默认100ms
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//      3、其他配置
//      props.put("linger.ms", 500);//延迟500ms发送，默认0
//      props.put("batch.size", 16384);//默认的批量处理消息字节数，默认16k
//      props.put("buffer.memory", 33554432);//producer可以用来缓存数据的内存大小，默认32M
//      props.put("max.request.size", 1048576);//producer每次发送的数据大小限制，默认1M
//      props.put("max.in.flight.requests.per.connection", 1);//该参数指定了生产者在收到服务器响应之前可以发送多少个消息，默认5，指定为1可防止消息乱序
        producer = new KafkaProducer<String, String>(props);
    }


}
